/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package pubsub

import (
	"github.com/golang/protobuf/proto"
	"hundsun.com/hsl/hschain/common/types"
	pbcom "hundsun.com/hsl/hschain/protos/common"
	"hundsun.com/hsl/hschain/store/db"
)

// Subscriber subscriber
type Subscriber struct {
	dbh db.Database
}

// NewSubscriber new subscriber
func NewSubscriber(dbm db.Database) *Subscriber {
	return &Subscriber{dbh: dbm}
}

// GetSubscriber get subscriber
func (p *Subscriber) GetSubscriber(ID string) (*pbcom.ReqSubscriber, error) {
	if p.dbh == nil {
		return nil, types.ErrNotFound
	}
	v, err := p.dbh.Get(genKey(ID))
	if err != nil {
		return nil, err
	}
	rsub := &pbcom.ReqSubscriber{}
	err = proto.Unmarshal(v, rsub)
	if err != nil {
		return nil, err
	}
	return rsub, nil
}

// GetEntireSubscribers get entire subscribers
func (p *Subscriber) GetEntireSubscribers() ([]*pbcom.ReqSubscriber, error) {
	if p.dbh == nil {
		return nil, types.ErrNotFound
	}
	lst := p.dbh.NewList()
	_, values, err := lst.ListPrefix([]byte(db.NamespaceSubscriber), -1, db.OptionWithValue)
	if err != nil {
		return nil, err
	}
	var reqs []*pbcom.ReqSubscriber
	for _, value := range values {
		req := &pbcom.ReqSubscriber{}
		err := proto.Unmarshal(value, req)
		if err != nil {
			return reqs, err
		}
		reqs = append(reqs, req)
	}
	return reqs, nil
}

// SubscriberKV ReqSubscriber to kv
func SubscriberKV(req *pbcom.ReqSubscriber) (*pbcom.KeyValue, error) {
	kv := &pbcom.KeyValue{}
	kv.Key = genKey(genSubscriberID(req))
	v, err := proto.Marshal(req)
	if err != nil {
		return nil, err
	}
	kv.Value = v
	return kv, nil
}

func genKey(ID string) []byte {
	return []byte(db.NamespaceSubscriber + ID)
}
